package cz.data.domain.market.mapping.listener;

import com.rabbitmq.client.Channel;
import cz.data.common.rabbitmq.config.RabbitConstant;
import cz.data.common.utils.ThrowableUtil;
import cz.data.domain.market.mapping.handler.DataApiHandlerMapping;
import cz.data.domain.market.mapping.handler.TableApiHandlerMapping;
import cz.data.domain.market.model.dto.ApiChangeMessage;
import cz.data.domain.market.model.entity.DataApiEntity;
import cz.data.domain.market.model.entity.TableApiEntity;
import cz.data.domain.market.rpc.DataApiServiceFeign;
import cz.data.domain.market.rpc.TableApiFeign;
import io.swagger.annotations.ApiOperation;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Objects;

@Slf4j
@Component
public class RabbitMqListener {

    private static final String HANDLER_RELEASE = "1";
    private static final String HANDLER_CANCEL = "2";

    @Autowired
    private DataApiServiceFeign dataApiServiceFeign;

    @Autowired
    private DataApiHandlerMapping dataApiHandlerMapping;
    @Resource
    TableApiFeign tableApiFeign;
    @Resource
    TableApiHandlerMapping tableApiHandlerMapping;

    @ApiOperation(value = "api发布与撤销监听", notes = "type 1:发布 2:撤销")
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(), //注意这里不要定义队列名称,系统会随机产生
            exchange = @Exchange(value = RabbitConstant.API_FANOUT_EXCHANGE, type = ExchangeTypes.FANOUT)
    ))
    public void fanoutQueueRelease(ApiChangeMessage apiChangeMessage, Channel channel, Message message) throws Exception {
        try {
            log.info("rabbit监听到API变更事件");
            if (ApiChangeMessage.DATA_API.equals(apiChangeMessage.getType())) {
                String id = apiChangeMessage.getId();
                String status = apiChangeMessage.getStatus();
                DataApiEntity dataApiEntity = dataApiServiceFeign.getDataApiById(id);
                if (dataApiEntity != null) {
                    log.info("监听到data api变更事件：{}-{},{}", id, dataApiEntity.getApiName(), status);
                    if (HANDLER_RELEASE.equals(status)) {
                        dataApiHandlerMapping.registerMapping(dataApiEntity);
                    } else if (HANDLER_CANCEL.equals(status)) {
                        dataApiHandlerMapping.unregisterMapping(dataApiEntity);
                    }
                }
            } else if (ApiChangeMessage.TABLE_API.equals(apiChangeMessage.getType())) {
                String id = apiChangeMessage.getId();
                String status = apiChangeMessage.getStatus();
                TableApiEntity tableApi = tableApiFeign.getTableApiById(id);
                if (Objects.nonNull(tableApi)) {
                    log.info("table api变更事件：{}-{},{}", id, tableApi.getApiName(), status);
                    if (HANDLER_RELEASE.equals(status)) {
                        tableApiHandlerMapping.registerMapping(tableApi);
                    } else if (HANDLER_CANCEL.equals(status)) {
                        tableApiHandlerMapping.unregisterMapping(tableApi);
                    }
                }
            }
        } catch (Exception e) {
            log.error("全局异常信息ex={}, StackTrace={}", e.getMessage(), ThrowableUtil.getStackTrace(e));
            if (message.getMessageProperties().getRedelivered()) {
                log.error("消息已处理,请勿重复处理！");
                // 拒绝消息
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
            } else {
                //记录日志
                log.error("消息消费失败处理：{}", e.getMessage());
                //第一个参数为消息的index，第二个参数是是否批量处理，第三个参数为是否让被拒绝的消息重新入队列
                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
            }
        } finally {
            // 手动确认
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }
    }

}
